package com.saber.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


/**
 * 共享的线程池
 * @author
 *
 */
public class SharedThreadPoolExecutor implements Executor {

	/**
	 * thread pool core size
	 */
	public static final int DEFAULT_CORE_SIZE = Runtime.getRuntime().availableProcessors() - 1;

	/**
	 * thread pool max size
	 */
	public static final int DEFAULT_MAX_SIZE = DEFAULT_CORE_SIZE * 2;

	/**
	 *
	 */
	public static final int DEAFAULT_KEEP_ALIVE_TIME = 60;

	/**
	 * work queue capacity
	 */
	public static final int DEFAULT_QUEUE_CAPACITY = 128;

	/**
	 * core size of thread pool
	 */
	private static int coreSize = DEFAULT_CORE_SIZE ;

	/**
	 * max size of thread pool
	 */
	private static int maxSize = DEFAULT_MAX_SIZE;

	/**
	 * keep alive time, unit second
	 */
	private static long keepAliveTime = DEAFAULT_KEEP_ALIVE_TIME;

	/**
	 * size of work queue capacity
	 */
	private static int workQueueCapacity = DEFAULT_QUEUE_CAPACITY;

	/**
	 * 工作队列
	 */
	private static BlockingQueue<Runnable> workQueue;

	/**
	 * 线程
	 */
	private static MyThreadPoolExecutor threadpool;

	/**
	 * 是否已经初始化
	 */
	private boolean isInitialized;

	/**
	 *
	 */
	private static Object[] static_lock = new Object[0];

	/**
	 *
	 */
	public void initialize() {
		synchronized(static_lock) {
			if(isInitialized) {
				return;
			}
			workQueue = new ArrayBlockingQueue<Runnable>(workQueueCapacity);
			threadpool = new MyThreadPoolExecutor(coreSize, maxSize,
					keepAliveTime, TimeUnit.SECONDS, workQueue,
					new NamedThreadFactory("shared-thread-pool"),
					new AlarmAndRetryRejectedExecutionHandler());

			isInitialized = true;
		}
	}

	/**
	 *
	 * @param command
	 */
	@Override
	public void execute(Runnable command) {
		threadpool.execute(command);
	}

	// Add By Liuwu Start
	public int getCorePoolSize(){
		return threadpool.getCorePoolSize();
	}
	public int getActiveCount(){
		return threadpool.getActiveCount();
	}

	public int getPoolSize(){
		return threadpool.getPoolSize();
	}

	public int getLargestPoolSize(){
		return threadpool.getLargestPoolSize();
	}

	public int getMaximumPoolSize(){
		return threadpool.getMaximumPoolSize();
	}

	public long getCompletedTaskCount(){
		return threadpool.getCompletedTaskCount();
	}

	public long getTaskCount(){
		return threadpool.getTaskCount();
	}

	public int getRemainingWorkQueue(){
		return workQueue.remainingCapacity();
	}
	// Add By Liuwu End

	/**
	 *
	 * @param coreSize
	 */
	public void setCoreSize(int coreSize) {
		SharedThreadPoolExecutor.coreSize = coreSize;
	}

	/**
	 *
	 * @param maxSize
	 */
	public void setMaxSize(int maxSize) {
		SharedThreadPoolExecutor.maxSize = maxSize;
	}

	/**
	 *
	 * @param keepAliveTime
	 */
	public void setKeepAliveTime(int keepAliveTime) {
		SharedThreadPoolExecutor.keepAliveTime = keepAliveTime;
	}

	/**
	 *
	 * @param workQueueCapacity
	 */
	public void setWorkQueueCapacity(int workQueueCapacity) {
		SharedThreadPoolExecutor.workQueueCapacity = workQueueCapacity;
	}

	/**
	 * 告警并重试的线程池繁忙拒绝服务的实现
	 *
	 * @author sihai
	 *
	 */
	public static class AlarmAndRetryRejectedExecutionHandler implements
			RejectedExecutionHandler {

		/**
		 * 默认最大重试次数
		 */
		public static final int DEFAULT_MAX_RETRY = 3;

		/**
		 * 默认提交失败休眠多长时间而后重试(单位毫秒)
		 */
		public static final int DEFAULT_SLEEP_TIME = 3;

		/**
		 * 最大重试次数
		 */
		private int maxRetry = DEFAULT_MAX_RETRY;

		/**
		 * 提交失败休眠多长时间而后重试(单位毫秒)
		 */
		private int sleepTime = DEFAULT_SLEEP_TIME;

		/**
		 * 使用默认最大重试次数和休眠时间构造方法
		 */
		public AlarmAndRetryRejectedExecutionHandler() {
			this(DEFAULT_MAX_RETRY, DEFAULT_SLEEP_TIME);
		}

		/**
		 * 使用指定最大重试次数构和指定的休眠时间造方法
		 *
		 * @param maxRetry
		 * @param sleepTime
		 */
		public AlarmAndRetryRejectedExecutionHandler(int maxRetry, int sleepTime) {
			this.maxRetry = maxRetry;
			this.sleepTime = sleepTime;
		}

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			// 1. 线程池被关闭了引起的reject
			if (executor.isShutdown()) {
				throw new RuntimeException("ThreadPool is shutdown");
			}

			// 2. TODO 告警

			// 3. 构造包装类，或者已经是包装类了
			RunnbaleWrapper rw = null;
			if (r instanceof RunnbaleWrapper) {
				rw = (RunnbaleWrapper) r;
			} else {
				rw = new RunnbaleWrapper(r, 0);
			}

			// 4. 判断是否已经超过了重试次数限制
//			if (rw.isRetryLimited()) {
//				throw new RuntimeException("Reach max retry count:" + this.maxRetry);
//			}

			// 5. 重试
			rw.incRetryed();
			this.retry(executor, rw);
		}

		/**
		 *
		 * @param executor
		 * @param rw
		 */
		private void retry(ThreadPoolExecutor executor, RunnbaleWrapper rw) {
			//
			try {
				Thread.sleep(this.sleepTime);
				//System.out.println("sharedThreadPool-> try:" + rw.retryedCount);
				executor.execute(rw);
			} catch (InterruptedException e) {
				// log
				Thread.currentThread().interrupt();
			}
		}

		/**
		 *
		 * @author
		 *
		 */
		private class RunnbaleWrapper implements Runnable {

			/**
			 *
			 */
			private Runnable target;

			/**
			 *
			 */
			private int retryedCount;

			/**
			 *
			 * @param target
			 * @param retryedCount
			 */
			public RunnbaleWrapper(Runnable target, int retryedCount) {
				this.target = target;
				this.retryedCount = retryedCount;
			}

			/**
			 * 记录重试
			 */
			public void incRetryed() {
				this.retryedCount++;
			}

			/**
			 * 是否达到重试次数限制了
			 *
			 * @return
			 */
			private boolean isRetryLimited() {
				return this.retryedCount >= AlarmAndRetryRejectedExecutionHandler.this.maxRetry;
			}

			@Override
			public void run() {
				if (null != target) {
					target.run();
				}
			}
		}
	}
}
